#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
MCP 基础工具示例

本文件展示了如何创建各种基础的 MCP 工具，包括：
- 字符串处理工具
- 数学计算工具
- 文件操作工具
- 系统信息工具
"""

from mcp.server.fastmcp import FastMCP
from typing import List, Dict, Optional, Union
import os
import math
import datetime
import json

# 创建 MCP 服务实例
mcp = FastMCP("Basic Tools Server")

# ============================================================================
# 字符串处理工具
# ============================================================================

@mcp.tool()
def text_transform(text: str, operation: str) -> str:
    """文本转换工具
    
    Args:
        text: 要转换的文本
        operation: 转换操作 (upper, lower, title, reverse, length)
        
    Returns:
        转换后的文本或文本信息
    """
    operations = {
        "upper": lambda t: t.upper(),
        "lower": lambda t: t.lower(),
        "title": lambda t: t.title(),
        "reverse": lambda t: t[::-1],
        "length": lambda t: f"文本长度: {len(t)} 字符"
    }
    
    if operation not in operations:
        return f"不支持的操作: {operation}. 支持的操作: {', '.join(operations.keys())}"
    
    try:
        result = operations[operation](text)
        return str(result)
    except Exception as e:
        return f"转换失败: {str(e)}"

@mcp.tool()
def text_search(text: str, pattern: str, case_sensitive: bool = False) -> str:
    """文本搜索工具
    
    Args:
        text: 要搜索的文本
        pattern: 搜索模式
        case_sensitive: 是否区分大小写
        
    Returns:
        搜索结果信息
    """
    try:
        search_text = text if case_sensitive else text.lower()
        search_pattern = pattern if case_sensitive else pattern.lower()
        
        count = search_text.count(search_pattern)
        
        if count == 0:
            return f"未找到模式 '{pattern}'"
        
        # 找到所有匹配位置
        positions = []
        start = 0
        while True:
            pos = search_text.find(search_pattern, start)
            if pos == -1:
                break
            positions.append(pos)
            start = pos + 1
        
        return f"找到 {count} 个匹配，位置: {positions}"
    except Exception as e:
        return f"搜索失败: {str(e)}"

@mcp.tool()
def text_replace(text: str, old: str, new: str, max_count: int = -1) -> str:
    """文本替换工具
    
    Args:
        text: 原始文本
        old: 要替换的文本
        new: 替换后的文本
        max_count: 最大替换次数，-1表示全部替换
        
    Returns:
        替换后的文本
    """
    try:
        if max_count == -1:
            result = text.replace(old, new)
        else:
            result = text.replace(old, new, max_count)
        
        original_count = text.count(old)
        actual_count = min(original_count, max_count) if max_count > 0 else original_count
        
        return f"替换完成，共替换 {actual_count} 处\n\n结果:\n{result}"
    except Exception as e:
        return f"替换失败: {str(e)}"

# ============================================================================
# 数学计算工具
# ============================================================================

@mcp.tool()
def basic_calculator(expression: str) -> str:
    """基础计算器
    
    Args:
        expression: 数学表达式（支持 +, -, *, /, **, (), sqrt, sin, cos, tan, log）
        
    Returns:
        计算结果
    """
    try:
        # 安全的数学函数
        safe_dict = {
            "__builtins__": {},
            "sqrt": math.sqrt,
            "sin": math.sin,
            "cos": math.cos,
            "tan": math.tan,
            "log": math.log,
            "log10": math.log10,
            "exp": math.exp,
            "pi": math.pi,
            "e": math.e
        }
        
        result = eval(expression, safe_dict)
        return f"表达式: {expression}\n结果: {result}"
    except ZeroDivisionError:
        return "错误: 除零错误"
    except ValueError as e:
        return f"数值错误: {str(e)}"
    except SyntaxError:
        return "语法错误: 表达式格式不正确"
    except Exception as e:
        return f"计算失败: {str(e)}"

@mcp.tool()
def number_info(number: float) -> str:
    """数字信息工具
    
    Args:
        number: 要分析的数字
        
    Returns:
        数字的详细信息
    """
    try:
        info = {
            "原始值": number,
            "绝对值": abs(number),
            "整数部分": int(number),
            "小数部分": number - int(number),
            "是否为整数": number.is_integer() if isinstance(number, float) else True,
            "是否为正数": number > 0,
            "是否为负数": number < 0,
            "平方": number ** 2,
            "平方根": math.sqrt(abs(number)) if number >= 0 else f"虚数: {math.sqrt(-number)}i"
        }
        
        result = "数字信息:\n"
        for key, value in info.items():
            result += f"{key}: {value}\n"
        
        return result
    except Exception as e:
        return f"分析失败: {str(e)}"

@mcp.tool()
def statistics_calculator(numbers: List[float]) -> str:
    """统计计算器
    
    Args:
        numbers: 数字列表
        
    Returns:
        统计信息
    """
    try:
        if not numbers:
            return "错误: 数字列表为空"
        
        n = len(numbers)
        total = sum(numbers)
        mean = total / n
        
        # 计算方差和标准差
        variance = sum((x - mean) ** 2 for x in numbers) / n
        std_dev = math.sqrt(variance)
        
        # 排序以计算中位数和四分位数
        sorted_numbers = sorted(numbers)
        
        # 中位数
        if n % 2 == 0:
            median = (sorted_numbers[n//2 - 1] + sorted_numbers[n//2]) / 2
        else:
            median = sorted_numbers[n//2]
        
        stats = {
            "数量": n,
            "总和": total,
            "平均值": mean,
            "中位数": median,
            "最小值": min(numbers),
            "最大值": max(numbers),
            "范围": max(numbers) - min(numbers),
            "方差": variance,
            "标准差": std_dev
        }
        
        result = "统计信息:\n"
        for key, value in stats.items():
            if isinstance(value, float):
                result += f"{key}: {value:.4f}\n"
            else:
                result += f"{key}: {value}\n"
        
        return result
    except Exception as e:
        return f"统计计算失败: {str(e)}"

# ============================================================================
# 文件操作工具
# ============================================================================

@mcp.tool()
def file_info(file_path: str) -> str:
    """文件信息工具
    
    Args:
        file_path: 文件路径
        
    Returns:
        文件的详细信息
    """
    try:
        if not os.path.exists(file_path):
            return f"文件不存在: {file_path}"
        
        stat = os.stat(file_path)
        
        info = {
            "文件路径": os.path.abspath(file_path),
            "文件名": os.path.basename(file_path),
            "目录": os.path.dirname(file_path),
            "文件大小": f"{stat.st_size} 字节",
            "是否为文件": os.path.isfile(file_path),
            "是否为目录": os.path.isdir(file_path),
            "创建时间": datetime.datetime.fromtimestamp(stat.st_ctime).strftime("%Y-%m-%d %H:%M:%S"),
            "修改时间": datetime.datetime.fromtimestamp(stat.st_mtime).strftime("%Y-%m-%d %H:%M:%S"),
            "访问时间": datetime.datetime.fromtimestamp(stat.st_atime).strftime("%Y-%m-%d %H:%M:%S")
        }
        
        # 如果是文件，添加扩展名信息
        if os.path.isfile(file_path):
            _, ext = os.path.splitext(file_path)
            info["文件扩展名"] = ext or "无扩展名"
        
        result = "文件信息:\n"
        for key, value in info.items():
            result += f"{key}: {value}\n"
        
        return result
    except Exception as e:
        return f"获取文件信息失败: {str(e)}"

@mcp.tool()
def directory_listing(directory: str, show_hidden: bool = False) -> str:
    """目录列表工具
    
    Args:
        directory: 目录路径
        show_hidden: 是否显示隐藏文件
        
    Returns:
        目录内容列表
    """
    try:
        if not os.path.exists(directory):
            return f"目录不存在: {directory}"
        
        if not os.path.isdir(directory):
            return f"路径不是目录: {directory}"
        
        items = os.listdir(directory)
        
        if not show_hidden:
            items = [item for item in items if not item.startswith('.')]
        
        items.sort()
        
        result = f"目录: {os.path.abspath(directory)}\n"
        result += f"项目数量: {len(items)}\n\n"
        
        for item in items:
            item_path = os.path.join(directory, item)
            if os.path.isdir(item_path):
                result += f"[DIR]  {item}/\n"
            else:
                size = os.path.getsize(item_path)
                result += f"[FILE] {item} ({size} 字节)\n"
        
        return result
    except Exception as e:
        return f"列出目录失败: {str(e)}"

@mcp.tool()
def text_file_reader(file_path: str, encoding: str = "utf-8", max_lines: int = 100) -> str:
    """文本文件读取工具
    
    Args:
        file_path: 文件路径
        encoding: 文件编码
        max_lines: 最大读取行数
        
    Returns:
        文件内容
    """
    try:
        if not os.path.exists(file_path):
            return f"文件不存在: {file_path}"
        
        if not os.path.isfile(file_path):
            return f"路径不是文件: {file_path}"
        
        with open(file_path, 'r', encoding=encoding) as f:
            lines = []
            for i, line in enumerate(f):
                if i >= max_lines:
                    break
                lines.append(line.rstrip('\n\r'))
        
        total_lines = len(lines)
        
        result = f"文件: {file_path}\n"
        result += f"编码: {encoding}\n"
        result += f"显示行数: {total_lines}\n"
        
        if total_lines >= max_lines:
            result += f"(文件可能有更多行，仅显示前 {max_lines} 行)\n"
        
        result += "\n" + "-" * 50 + "\n"
        
        for i, line in enumerate(lines, 1):
            result += f"{i:4d}: {line}\n"
        
        return result
    except UnicodeDecodeError:
        return f"编码错误: 无法使用 {encoding} 编码读取文件"
    except Exception as e:
        return f"读取文件失败: {str(e)}"

# ============================================================================
# 系统信息工具
# ============================================================================

@mcp.tool()
def current_time(format_string: str = "%Y-%m-%d %H:%M:%S") -> str:
    """当前时间工具
    
    Args:
        format_string: 时间格式字符串
        
    Returns:
        格式化的当前时间
    """
    try:
        now = datetime.datetime.now()
        formatted_time = now.strftime(format_string)
        
        result = f"当前时间: {formatted_time}\n"
        result += f"时间戳: {now.timestamp()}\n"
        result += f"星期: {now.strftime('%A')}\n"
        result += f"年份第几天: {now.timetuple().tm_yday}\n"
        
        return result
    except Exception as e:
        return f"获取时间失败: {str(e)}"

@mcp.tool()
def environment_info() -> str:
    """环境信息工具
    
    Returns:
        系统环境信息
    """
    try:
        import platform
        
        info = {
            "操作系统": platform.system(),
            "系统版本": platform.version(),
            "处理器架构": platform.machine(),
            "Python版本": platform.python_version(),
            "当前工作目录": os.getcwd(),
            "用户主目录": os.path.expanduser("~"),
            "临时目录": os.path.dirname(os.path.abspath(__file__))
        }
        
        result = "环境信息:\n"
        for key, value in info.items():
            result += f"{key}: {value}\n"
        
        # 添加一些环境变量
        important_env_vars = ['PATH', 'HOME', 'USER', 'PYTHONPATH']
        result += "\n重要环境变量:\n"
        for var in important_env_vars:
            value = os.environ.get(var, "未设置")
            if var == 'PATH':
                # PATH 变量太长，只显示前几个路径
                paths = value.split(os.pathsep)[:3]
                value = os.pathsep.join(paths) + "..."
            result += f"{var}: {value}\n"
        
        return result
    except Exception as e:
        return f"获取环境信息失败: {str(e)}"

# ============================================================================
# 数据格式转换工具
# ============================================================================

@mcp.tool()
def json_formatter(json_string: str, indent: int = 2) -> str:
    """JSON 格式化工具
    
    Args:
        json_string: JSON 字符串
        indent: 缩进空格数
        
    Returns:
        格式化后的 JSON
    """
    try:
        data = json.loads(json_string)
        formatted = json.dumps(data, indent=indent, ensure_ascii=False, sort_keys=True)
        
        result = "格式化后的 JSON:\n"
        result += "-" * 30 + "\n"
        result += formatted
        
        return result
    except json.JSONDecodeError as e:
        return f"JSON 解析错误: {str(e)}"
    except Exception as e:
        return f"格式化失败: {str(e)}"

@mcp.tool()
def data_converter(data: str, from_format: str, to_format: str) -> str:
    """数据格式转换工具
    
    Args:
        data: 原始数据
        from_format: 源格式 (json, csv)
        to_format: 目标格式 (json, csv, table)
        
    Returns:
        转换后的数据
    """
    try:
        # 解析源数据
        if from_format.lower() == "json":
            parsed_data = json.loads(data)
        elif from_format.lower() == "csv":
            import csv
            import io
            
            reader = csv.DictReader(io.StringIO(data))
            parsed_data = list(reader)
        else:
            return f"不支持的源格式: {from_format}"
        
        # 转换为目标格式
        if to_format.lower() == "json":
            result = json.dumps(parsed_data, indent=2, ensure_ascii=False)
        elif to_format.lower() == "csv":
            if not isinstance(parsed_data, list) or not parsed_data:
                return "CSV 转换需要非空的对象列表"
            
            import csv
            import io
            
            output = io.StringIO()
            writer = csv.DictWriter(output, fieldnames=parsed_data[0].keys())
            writer.writeheader()
            writer.writerows(parsed_data)
            result = output.getvalue()
        elif to_format.lower() == "table":
            if not isinstance(parsed_data, list) or not parsed_data:
                return "表格转换需要非空的对象列表"
            
            headers = list(parsed_data[0].keys())
            
            # 计算列宽
            col_widths = {}
            for header in headers:
                col_widths[header] = max(
                    len(str(header)),
                    max(len(str(row.get(header, ""))) for row in parsed_data)
                )
            
            # 生成表格
            result = ""
            
            # 表头
            header_row = " | ".join(str(h).ljust(col_widths[h]) for h in headers)
            result += header_row + "\n"
            
            # 分隔线
            separator = "-+-".join("-" * col_widths[h] for h in headers)
            result += separator + "\n"
            
            # 数据行
            for row in parsed_data:
                data_row = " | ".join(
                    str(row.get(h, "")).ljust(col_widths[h]) for h in headers
                )
                result += data_row + "\n"
        else:
            return f"不支持的目标格式: {to_format}"
        
        return f"转换完成 ({from_format} -> {to_format}):\n\n{result}"
    except Exception as e:
        return f"转换失败: {str(e)}"

# ============================================================================
# 启动服务
# ============================================================================

if __name__ == "__main__":
    mcp.run(transport="stdio")